package com.yhb.service.impl;

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.TopicMessage;
import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.yhb.model.MQMessage;
import com.yhb.model.mq.*;
import com.yhb.service.*;
import com.yhb.third.leshua.LeShuaService;
import com.yhb.yinsheng.UnionPayService;
import lombok.extern.log4j.Log4j;
import lombok.var;
import me.edzh.spring.bootstrap.utils.Constants;
import me.edzh.spring.bootstrap.utils.JSONUtil;
import me.edzh.spring.bootstrap.utils.ResultCode;
import me.edzh.spring.bootstrap.utils.exception.ServiceException;
import org.apache.log4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;


@Log4j
@Service
@EnableAsync
public class MQServiceImpl implements MessageListener, MQService {

    @Autowired(required = false)
    private Producer producer;

    @Value("${aliyun.mq.topic}")
    private String topic;

    @Value("${aliyun.mq.tag}")
    private String tag;

    @Value("${aliyun.mq.mode}")
    private String mqMode;

    private static MQProducer httpProducer;

    private static MQConsumer httpConsumer;

    private static MQClient httpMqClient;

    @Value("${aliyun.mq.gid}")
    private String mqGid;

    @Value("${aliyun.mq.accessPoint}")
    private String mqAccessPoint;

    @Value("${aliyun.hignend.ak}")
    private String ak;

    @Value("${aliyun.hignend.sk}")
    private String sk;


    @Autowired
    private TradeService tradeService;

    @Autowired
    private UnionPayService unionPayService;

    @Autowired
    private SplittingService splittingService;

    @Autowired
    private LeShuaService leShuaService;

    @Autowired
    private MPosService mPosService;

    @Autowired
    private UserService userService;

    @Autowired
    private CpsService cpsService;

    @Autowired
    private ChannelEventBusImpl channelEventBus;

    @PostConstruct
    private void init() {
        if ("local".equals(mqMode) && httpMqClient == null) {
            httpMqClient = new MQClient(mqAccessPoint, ak, sk);
            httpProducer = httpMqClient.getProducer("MQ_INST_1825751410379534_BbeEsIF8", topic);
            httpConsumer = httpMqClient.getConsumer("MQ_INST_1825751410379534_BbeEsIF8", topic, mqGid, tag);
            log.info("start listen:" + Thread.currentThread().getName());

            startListen();
            log.info("mq http init");
        }
    }

    @Async
    @Override
    public void startListen() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                log.info("start listen:" + Thread.currentThread().getName());

                for (;;) {
                    List<com.aliyun.mq.http.model.Message> messages = null;

                    try {
                        // 长轮询消费消息
                        // 长轮询表示如果topic没有消息则请求会在服务端挂住3s，3s内如果有消息可以消费则立即返回
                        messages = httpConsumer.consumeMessage(
                                3,// 一次最多消费3条(最多可设置为16条)
                                10// 长轮询时间3秒（最多可设置为30秒）
                        );
                    } catch (Throwable e) {
                        log.error(e);
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e1) {
                            log.error(e1);
                        }
                    }
                    // 没有消息
                    if (messages == null || messages.isEmpty()) {
                        log.info(Thread.currentThread().getName() + ": no new message, continue!");
                        continue;
                    }

                    // 处理业务逻辑
                    for (com.aliyun.mq.http.model.Message message : messages) {
                        log.info("Receive message: " + message);

                    }

                    // Message.nextConsumeTime前若不确认消息消费成功，则消息会重复消费
                    // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样
                    {
                        List<String> handles = new ArrayList<String>();
                        for (com.aliyun.mq.http.model.Message message : messages) {
                            String content = new String(message.getMessageBodyBytes(), StandardCharsets.UTF_8);

                            log.info("received message:" + "  "  + message.getMessageId()  + " tag: " + message.getMessageTag() + " content:" + content);

                            consumeMsg(topic, message.getMessageKey(), message.getMessageTag(), message.getMessageBodyBytes());
                            handles.add(message.getReceiptHandle());
                        }

                        try {
                            httpConsumer.ackMessage(handles);
                        } catch (Throwable e) {
                            // 某些消息的句柄可能超时了会导致确认不成功
                            if (e instanceof AckMessageException) {
                                AckMessageException errors = (AckMessageException) e;
                                log.error("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                                if (errors.getErrorMessages() != null) {
                                    for (String errorHandle :errors.getErrorMessages().keySet()) {
                                        log.error("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                                + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                                    }
                                }
                                continue;
                            }
                            log.error(e);
                        }
                    }
                }
            }
        }, "consumer").start();

    }

    @PreDestroy
    private void shutdown() {
        if (httpMqClient != null) {
            httpMqClient.close();
            httpMqClient = null;
        }
    }



    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        MDC.put(Constants.LOG_TRACE_ID, UUID.randomUUID().toString()); // 入口处设置uuid

        String content = new String(message.getBody(), StandardCharsets.UTF_8);

        log.info("received message:" + "  " + message.getTopic() + "  " + message.getMsgID() + "tag: " + message.getTag() + " content:" + content);

        consumeMsg(message.getTopic(), message.getKey(), message.getTag(), message.getBody());
        MDC.clear();
        return Action.CommitMessage;
    }


    @Async
    protected void consumeMsg(String topic, String key, String tag, byte[] body) {
        try {
            MQMessage payload = MQMessage.fromJson(body);
            log.info("received mq:" + payload.getType());
            switch (payload.getType()) {
                case MQMessage.T_CALC_FLOW_SUM_UP: { // 计算日交易额
                    splittingService.sumUpAmount((SumUpParam) payload.getContent());
                    break;
                }
                case MQMessage.T_CALC_MONTHLY_FLOW_SUM_UP: { // 计算月交易额
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.calculateMonthlyAmount(param.getMonthString(), false);
                    break;
                }
                case MQMessage.T_START_CALC_SPLITTINGS: { // 计算月交易额后开始分润计算
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.calculateMonthlyAmount(param.getMonthString(), true);
                    break;
                }
                case MQMessage.T_CALC_FLOW_SPLITTINGS: { // 计算某日所有交易记录对应分润
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.sumUpSplitting(param.getDateString(), param.getMonthString());
                    break;
                }
                case MQMessage.T_START_OUTPUT_INCOME_REPORT: {
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.startOutputIncomeReport(param.getMonthString());
                    break;
                }
                case MQMessage.T_START_OUTPUT_INCOME_SUMMARY_REPORT: {
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.startOutputIncomeSummaryReport(param.getMonthString());
                    break;
                }
                case MQMessage.T_CALC_MONTHLY_ACTIVATE_BONUS: {
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.calculateMonthlyActivateBonus(param.getMonthString(), param.getUserId());
                    break;
                }
                case MQMessage.T_OUTPUT_INCOME_REPORT: {
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.outputIncomeReport(param.getMonthString(), param.getUserId());
                    break;
                }
                case MQMessage.T_QUERY_UNION_PAY_ORDER: { // 查询银盛订单详情
                    QueryUnionPayOrderParam param = (QueryUnionPayOrderParam) payload.getContent();
                    unionPayService.queryOrder(param.getOrderId());
                    break;
                }
                case MQMessage.T_CHECK_VIP_VALID: { // 检查vip状态, 并扭转
                    SumUpParam param = (SumUpParam) payload.getContent();
                    userService.checkVipValidation(param.getUserId());
                    break;
                }

                case MQMessage.T_SYNC_LESHUA_MERCHANTS: { // 同步乐刷商户类型
                    TimeRangeParam param = (TimeRangeParam) payload.getContent();
                    mPosService.syncMerchantId(param.getStartTime(), param.getEndTime());
                    break;
                }
                case MQMessage.T_SYNC_LESHUA_ORDERS: { // 同步乐刷订单
                    TimeRangeParam param = (TimeRangeParam) payload.getContent();
                    mPosService.syncOrders(param.getStartTime(), param.getEndTime());
                    break;
                }
                case MQMessage.T_OUTPUT_INCOME_SUMMARY_REPORT: { // 生成summary
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.outputIncomeSummaryReport(param.getMonthString(), param.getUserId());
                    break;
                }
                case MQMessage.T_SUM_UP_APPROXIMATELY_INCOME: {
                    SumUpParam param = (SumUpParam) payload.getContent();
                    splittingService.sumUpApproximatelyIncome(param.getUserId(), param.getDateString());
                    break;
                }
                case MQMessage.T_CALC_ACTIVE_USER: {
                    TimeRangeParam param = (TimeRangeParam) payload.getContent();
                    cpsService.calcTreeActiveCount(param.getUserId(), param.getStartTime(), param.getEndTime());
                    break;
                }
                case MQMessage.T_CHANNEL_EVENT_BUS: {
                    ChannelEvent e = (ChannelEvent) payload.getContent();
                    channelEventBus.onEvent(e.getPayload());
                    break;
                }
                case MQMessage.T_QUERY_UNION_PAY_ACTIVE_REMIT: {
                    QueryUnionFlowParam p = (QueryUnionFlowParam) payload.getContent();
                    unionPayService.queryFlow(p.getId(), p.getDateString());
                    break;
                }
                case MQMessage.T_CLEAR_USER_RESERVED_BALANCE: {//日终清分队列
                    SumUpParam param = (SumUpParam) payload.getContent();
                    var user = userService.getUser(param.getUserId());
                    // YS通道下架
                    // unionPayService.clearUserBalance(user);
                    tradeService.clearUserReservedBalance(user);
                    break;
                }

                case MQMessage.T_CHECK_INVITEE_CERTIFICATE: {
                    TimeRangeParam param = (TimeRangeParam) payload.getContent();
                    var user = userService.getUser(param.getUserId());
                    userService.checkInvitorVIPBonusMonthly(user, param.getStartTime(), param.getEndTime());
                    break;
                }

                case MQMessage.T_CALLBACK_THIRD_FLOW: {
                    IdParam param = (IdParam) payload.getContent();
                    tradeService.checkThirdFlow(param.getDateString(), param.getId());
                    break;
                }
                default: {
                    throw new ServiceException(ResultCode.ERROR_PARAMETER_WRONG, "unknown message type");
                }
            }
        } catch (Throwable t) {
            log.error("", t);
        }

    }

    @Async
    @Override
    public void sendMsg(MQMessage payload) {
        payload.setHeader(MQMessage.MQHeader.builder().relatedUserId((String)MDC.get(Constants.LOG_USER_ID)).build());

        Message msg = new Message(topic, tag, JSONUtil.toJson(payload).getBytes(StandardCharsets.UTF_8));
        String msgKey = UUID.randomUUID().toString();
        msg.setKey(msgKey);

        if (httpProducer != null) { // http模式. 不另外发
            TopicMessage message = new TopicMessage(
                    msg.getBody(),
                    tag
            );

            message.setMessageKey(msgKey);
            try {
                TopicMessage result = httpProducer.publishMessage(message);
                log.info("send msg success, id: " + result.getMessageId() + " key: " + msgKey + " tag: " + tag + " type:" + payload.getType());
            } catch (ONSClientException e) {
                log.warn("send msg err", e);
            }

            return;
        }

        try {
            SendResult result = producer.send(msg);
            assert result != null;
            log.info("send msg success, id: " + result.getMessageId() + " key: " + msgKey + " tag: " + tag + " type:" + payload.getType());
        } catch (ONSClientException e) {
            log.warn("send msg err", e);
        }
    }



}
